package operators;

import entities.*;
import java.util.*;
import metadata.*;

/**
 *
 * @author Dragan
 */
public class QueryExecutionPlan {

    private Domain domain;
    private Query query;
    private ViewGene view;
    private long diskCost;
    SystemMetaData smd;
    OperationsHolder executionPlan;

    public QueryExecutionPlan(Domain domain, Query query, ViewGene view) {
        this.domain = domain;
        this.query = query;
        this.view = view;
        smd = domain.getSystemMetaData();
    }

    // TODO parallel operations
    public OperationsHolder executePlan() {
        OperationsHolder<Operation> execPlan = new OperationsHolder<Operation>();
        execPlan.addAll(executeSelection());
        execPlan.addAll(joinAll(execPlan.get(execPlan.size() - 1).getOutputSize()));
        diskCost = 0;
        for (Operation o : execPlan) {
            diskCost += o.getDiskCost();
        }
        this.executionPlan = execPlan;
        return execPlan;
    }

    public OperationsHolder getExecutionPlan() {
        return executionPlan;
    }

    OperationsHolder<Operation> executeSelection() {
        int whereSize = query.getWhere().size();
        OperationsHolder<Operation> operations = new OperationsHolder<Operation>();
        TableMetaData viewMetaData = view.getTableMetaData(smd);
        if (whereSize == 0) {
            // Scan through all the tuples
            operations.add(new TableScan(viewMetaData, viewMetaData.getNumberOfTuples(), 1.0));
        } else if (whereSize == 1) {
            Table dimTable = query.getWhere().get(0).getAtribute().getTable();
            operations.addAll(this.selectionOnOneTable(dimTable, query.getWhere()));
        } else if (whereSize > 1) {
            HashMap<Table, List<Condition>> relationConditions = new HashMap<Table, List<Condition>>();
            for (Condition c : query.getWhere()) {
                if (!relationConditions.containsKey(c.getAtribute().getTable())) {
                    relationConditions.put(c.getAtribute().getTable(), new ArrayList<Condition>());
                }
                relationConditions.get(c.getAtribute().getTable()).add(c);
            }
            int relCount = relationConditions.keySet().size();
            TableMetaData[] tables = new TableMetaData[relCount];
            long[] inputs = new long[relCount];
            int i = 0;
            for (Table dimensionRelation : relationConditions.keySet()) {
                OperationsHolder<Operation> ops = selectionOnOneTable(dimensionRelation, relationConditions.get(dimensionRelation));
                tables[i] = dimensionRelation.getTableMetaData(smd);
                inputs[i] = ops.get(ops.size() - 1).getOutputSize();
                i++;
                operations.addAll(ops);
            }
            Operation tidIntersection = new TidIntersection(viewMetaData, tables, inputs);
            Operation tableAccess = new TableAccess(viewMetaData, tidIntersection.getOutputSize());
            operations.add(tidIntersection);
            operations.add(tableAccess);
        }
        return operations;
    }

    OperationsHolder<Operation> selectionOnOneTable(Table dimensionTable, List<Condition> conditions) {
        OperationsHolder<Operation> operations = new OperationsHolder<Operation>();
        TableMetaData viewMetaData = view.getTableMetaData(smd);
        List<Atribute> indexedAtributes = new ArrayList<Atribute>();
        boolean doJoin = false;
        for (Condition c : conditions) {
            if (!view.isAtributeMaterialized(c.getAtribute())) {
                doJoin = true;
            } else {
                if (view.isAtributeIndexed(c.getAtribute())) {
                    indexedAtributes.add(c.getAtribute());
                }
            }
        }
        if (doJoin) {
            operations.addAll(tableAccess(dimensionTable, conditions));
            long dimTableAccessOutputSize = operations.get(operations.size() - 1).getOutputSize();
            Atribute pkATribute = dimensionTable.getPrimaryKey().getAtribute();
            if (view.isAtributeIndexed(pkATribute)) {
                Operation indexAccess = new IndexAccess(viewMetaData, view.getIndex(pkATribute).getIndexMetaData(smd.getDiskPageSizeInBytes()), dimTableAccessOutputSize);
                Operation viewTableAccess = new TableAccess(viewMetaData, indexAccess.getOutputSize());
                operations.add(indexAccess);
                operations.add(viewTableAccess);
            } else {
                TableScan viewScan = new TableScan(viewMetaData, view.getNumberOfTuples());
                TableMetaData t1 = dimensionTable.getTableMetaData(smd);
                TableMetaData t2 = viewMetaData;
                HashJoin hashJoin = new HashJoin(smd, t1, t2, dimTableAccessOutputSize, viewScan.getOutputSize());
                operations.addParallelToPrevious(viewScan);
                operations.add(hashJoin);
            }
        } else {
            operations.addAll(tableAccess(view, conditions));
        }
        return operations;
    }

    OperationsHolder<Operation> tableAccess(Table table, List<Condition> conditions) {
        OperationsHolder<Operation> operations = new OperationsHolder<Operation>();
        // 1.0 means total selection of all tuplese in table
        double totalSelectivity = 1.0;
        List<Operation> indexScans = new ArrayList<Operation>();
        for (Condition c : conditions) {
            double sel = c.getSelectivity();
            totalSelectivity *= sel;
            Atribute a = c.getAtribute();
            if (table.isAtributeIndexed(a)) {
                IndexScan is = new IndexScan(smd, table.getTableMetaData(smd), table.getIndex(a).getIndexMetaData(smd.getDiskPageSizeInBytes()), table.getNumberOfTuples(), sel);
                indexScans.add(is);
            }
        }
        int numberOfIndexedAtributes = indexScans.size();
        if (numberOfIndexedAtributes > 0) {
            operations.addAllParallel(indexScans);
        }
        if (numberOfIndexedAtributes == 1) {
            TableAccess ta = new TableAccess(table.getTableMetaData(smd), indexScans.get(0).getOutputSize());
            operations.add(ta);
        } else if (numberOfIndexedAtributes > 1) {
            TableMetaData[] tables = new TableMetaData[numberOfIndexedAtributes];
            long[] inputs = new long[numberOfIndexedAtributes];
            TableMetaData tmd = table.getTableMetaData(smd);
            for (int i = 0; i < numberOfIndexedAtributes; i++) {
                tables[i] = tmd;
                inputs[i] = indexScans.get(i).getOutputSize();
            }
            TidIntersection tid = new TidIntersection(tmd, tables, inputs);
            TableAccess ta = new TableAccess(table.getTableMetaData(smd), indexScans.get(0).getOutputSize());
            operations.addAllParallel(indexScans);
            operations.add(tid);
            operations.add(ta);
        } else if (numberOfIndexedAtributes == 0) {
            TableMetaData tmd = table.getTableMetaData(domain.getSystemMetaData());
            long numberOfTuples = tmd.getNumberOfTuples();
            // We must select at least one tuple from table
            if (totalSelectivity * numberOfTuples < 1) {
                totalSelectivity = 1 / (double) numberOfTuples;
            }
            operations.add(new TableScan(tmd, numberOfTuples, totalSelectivity));
        }
        return operations;
    }

    OperationsHolder<Operation> dimensionTableAccess(Table table) {
        return tableAccess(table, new ArrayList<Condition>());
    }

    public OperationsHolder<Operation> joinAll(long inputSize) {
        List<Table> relationsToJoin = new ArrayList<Table>();
        for (Atribute atr : query.getGroupby()) {
            // If the atribute is not denormalized then perform a join
            if (!view.isAtributeMaterialized(atr)) {
                if (!relationsToJoin.contains(atr.getTable())) {
                    relationsToJoin.add(atr.getTable());
                }
            }
        }
        OperationsHolder<Operation> operations = new OperationsHolder<Operation>();
        OperationsHolder<Operation> ops = new OperationsHolder<Operation>();
        List<Table> joinedRelations = new ArrayList<Table>();
        for (Table r : relationsToJoin) {
            // r could be added by a previous join with a join path
            if (!joinedRelations.contains(r)) {
                joinedRelations.addAll(join(view, r, inputSize, ops));
            }
        }
        // TODO treba addAllParallel
        operations.addAll(ops);
        return operations;
    }

    // Ne e dobro 
    public List<Table> join(ViewGene view, Table relation, long inputSize, OperationsHolder<Operation> operations) {
        List<Table> joinPath = findNearestMatParent(view, relation, new ArrayList<Table>());
        // relation is always the first Table on the join path
        Table joinTable1 = view;
        for (Table joinTable2 : joinPath) {
            Atribute pk = joinTable2.getPrimaryKey().getAtribute();
            if (joinTable2.isAtributeIndexed(pk)) {
                IndexAccess ia = new IndexAccess(joinTable2.getTableMetaData(smd), joinTable2.getIndex(pk).getIndexMetaData(smd.getDiskPageSizeInBytes()), inputSize);
                TableAccess ta = new TableAccess(joinTable2.getTableMetaData(smd), ia.getOutputSize());
                operations.add(ia);
                operations.add(ta);
            } else {
                TableScan tsDimTable = new TableScan(joinTable2.getTableMetaData(smd), joinTable2.getNumberOfTuples());
                TableScan tsFactTable = new TableScan(joinTable1.getTableMetaData(smd), inputSize);
                HashJoin hj = new HashJoin(smd, joinTable2.getTableMetaData(smd), joinTable1.getTableMetaData(smd), tsDimTable.getOutputSize(), inputSize);
                operations.add(tsDimTable);
                operations.add(tsFactTable);
                operations.add(hj);
            }
            joinTable1 = joinTable2;
        }
        return joinPath;
    }

    // TODO go vraka mat parent po depth first, a treba breadth first za da se dobie nakratkiot pat
    // od parent do relation
    private List<Table> findNearestMatParent(ViewGene view, Table relation, List<Table> parentPath) {
        parentPath.add(relation);
        if (view.isAtributeMaterialized(relation.getPrimaryKey().getAtribute())) {
            return parentPath;
        }
        for (Table parent : relation.getParents()) {
            if (view.isAtributeMaterialized(parent.getPrimaryKey().getAtribute())) {
                parentPath.add(parent);
                return parentPath;
            } else {
                List<Table> parents = findNearestMatParent(view, parent, new ArrayList<Table>(parentPath));
                if (parents != null) {
                    return parents;
                }
            }
        }
        return null;
    }

    public long getDiskCost() {
        return diskCost;
    }

    @Override
    public String toString() {
        String str = "diskCost:" + diskCost + " = " + executionPlan.getShortDesc();
        str += "\nview:" + view.getBasicDesc();
        str += "\nquery:" + query.toString();
        return str;
    }
}
